package org.logsin37.tool.eurekamanager.domain.service.impl;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import org.logsin37.tool.eurekamanager.domain.entity.FrozenConfig;
import org.logsin37.tool.eurekamanager.domain.service.EurekaRemoteDomainService;
import org.logsin37.tool.eurekamanager.domain.service.FrozenDomainService;
import org.logsin37.tool.eurekamanager.infra.constant.ErrorMessages;
import org.logsin37.tool.eurekamanager.infra.repository.FrozenConfigRepository;
import org.logsin37.tool.eurekamanager.infra.task.runner.FrozenTaskRunner;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
 * Frozen Domain Service Implement
 *
 * @author Greco 2021-11-23
 */
@Service
public class FrozenDomainServiceImpl implements FrozenDomainService, InitializingBean {

    private final FrozenConfigRepository frozenConfigRepository;
    private final EurekaRemoteDomainService eurekaRemoteDomainService;

    private final ConcurrentHashSet<String> frozenSet = new ConcurrentHashSet<>();

    public FrozenDomainServiceImpl(
            FrozenConfigRepository frozenConfigRepository,
            EurekaRemoteDomainService eurekaRemoteDomainService
    ) {
        this.frozenConfigRepository = frozenConfigRepository;
        this.eurekaRemoteDomainService = eurekaRemoteDomainService;
    }

    @Override
    public void afterPropertiesSet() {
        this.frozenConfigRepository.listFrozenConfig()
                .publishOn(Schedulers.elastic())
                .subscribeOn(Schedulers.elastic())
                .map(FrozenConfig::getEurekaCode)
                .filter(eurekaCode -> Boolean.TRUE.equals(this.frozenConfigRepository.isFrozen(eurekaCode).block()))
                .flatMap(this::frozen)
                .subscribe();
    }

    @Override
    public Mono<Void> frozen(String eurekaCode) {
        // 校验参数
        Assert.notNull(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        // 如果当前Eureka代码已冻结则不执行
        if(this.frozenSet.contains(eurekaCode)) {
            return Mono.empty();
        }
        return Mono.just(eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                // 根据code获取冻结配置
                .flatMap(this.frozenConfigRepository::getFrozenConfigByEurekaCode)
                // 中间参数校验
                .doOnNext(frozenConfig -> Assert.notNull(frozenConfig, StrUtil.format("can not find frozen config by code {}", eurekaCode)))
                .doOnNext(frozenConfig -> {
                    // 内存中存储冻结记录
                    this.frozenSet.add(eurekaCode);
                    // 持久化冻结记录
                    this.frozenConfigRepository.saveFrozenRecord(eurekaCode).subscribe();
                })
                // 新建一个线程来循环执行锁定任务
                .doOnNext(frozenConfig ->
                    ThreadUtil.execute(new FrozenTaskRunner(
                            frozenConfig,
                            this.frozenConfigRepository,
                            this.eurekaRemoteDomainService,
                            this
                    ))
                )
                .then();
    }

    @Override
    public Mono<Void> unfrozen(String eurekaCode) {
        Assert.notNull(eurekaCode, String.format(ErrorMessages.NOT_NULL, "eureka code"));
        return Mono.just(eurekaCode)
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .flatMap(this.frozenConfigRepository::getFrozenConfigByEurekaCode)
                .doOnNext(frozenConfig -> Assert.notNull(frozenConfig, StrUtil.format("can not find frozen config by code {}", eurekaCode)))
                .doOnNext(frozenConfig -> this.frozenSet.remove(eurekaCode))
                .flatMap(frozenConfig -> this.frozenConfigRepository.dropFrozenRecord(frozenConfig.getEurekaCode()));
    }
}
